/**
 * Software License, Version 1.0
 * 
 * Copyright 2003 The Trustees of Indiana University. All rights reserved.
 * 
 * 
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are met:
 * 
 * 1) All redistributions of source code must retain the above copyright notice,
 * the list of authors in the original source code, this list of conditions and
 * the disclaimer listed in this license; 2) All redistributions in binary form
 * must reproduce the above copyright notice, this list of conditions and the
 * disclaimer listed in this license in the documentation and/or other materials
 * provided with the distribution; 3) Any documentation included with all
 * redistributions must include the following acknowledgement:
 * 
 * "This product includes software developed by the Community Grids Lab. For
 * further information contact the Community Grids Lab at
 * http://communitygrids.iu.edu/."
 * 
 * Alternatively, this acknowledgement may appear in the software itself, and
 * wherever such third-party acknowledgments normally appear.
 * 
 * 4) The name Indiana University or Community Grids Lab or NaradaBrokering,
 * shall not be used to endorse or promote products derived from this software
 * without prior written permission from Indiana University. For written
 * permission, please contact the Advanced Research and Technology Institute
 * ("ARTI") at 351 West 10th Street, Indianapolis, Indiana 46202. 5) Products
 * derived from this software may not be called NaradaBrokering, nor may Indiana
 * University or Community Grids Lab or NaradaBrokering appear in their name,
 * without prior written permission of ARTI.
 * 
 * 
 * Indiana University provides no reassurances that the source code provided
 * does not infringe the patent or any other intellectual property rights of any
 * other entity. Indiana University disclaims any liability to any recipient for
 * claims brought by any other entity based on infringement of intellectual
 * property rights or otherwise.
 * 
 * LICENSEE UNDERSTANDS THAT SOFTWARE IS PROVIDED "AS IS" FOR WHICH NO
 * WARRANTIES AS TO CAPABILITIES OR ACCURACY ARE MADE. INDIANA UNIVERSITY GIVES
 * NO WARRANTIES AND MAKES NO REPRESENTATION THAT SOFTWARE IS FREE OF
 * INFRINGEMENT OF THIRD PARTY PATENT, COPYRIGHT, OR OTHER PROPRIETARY RIGHTS.
 * INDIANA UNIVERSITY MAKES NO WARRANTIES THAT SOFTWARE IS FREE FROM "BUGS",
 * "VIRUSES", "TROJAN HORSES", "TRAP DOORS", "WORMS", OR OTHER HARMFUL CODE.
 * LICENSEE ASSUMES THE ENTIRE RISK AS TO THE PERFORMANCE OF SOFTWARE AND/OR
 * ASSOCIATED MATERIALS, AND TO THE PERFORMANCE AND VALIDITY OF INFORMATION
 * GENERATED USING SOFTWARE.
 */
package cgl.narada.service.qos.impl;

import java.security.PublicKey;
import java.security.cert.Certificate;
import java.util.Enumeration;
import java.util.Hashtable;
import java.util.Properties;
import java.util.Vector;

import javax.crypto.SecretKey;

import cgl.narada.benchmark.security.BenchmarkListener;
import cgl.narada.event.EventHeaders;
import cgl.narada.event.NBEvent;
import cgl.narada.event.NBEventException;
import cgl.narada.event.NBEventTypes;
import cgl.narada.event.impl.NBEventGenerator;
import cgl.narada.matching.EntityProfileMatchingTrees;
import cgl.narada.matching.Profile;
import cgl.narada.matching.ProfileRequest;
import cgl.narada.service.ServiceException;
import cgl.narada.service.ServiceProperties;
import cgl.narada.service.buffering.BufferingServiceImpl;
import cgl.narada.service.client.ConnectionLossListener;
import cgl.narada.service.client.EventConsumer;
import cgl.narada.service.client.EventProducer;
import cgl.narada.service.client.impl.EventConsumerImpl;
import cgl.narada.service.client.impl.EventProducerImpl;
import cgl.narada.service.compression.CompressionService;
import cgl.narada.service.compression.DecompressionService;
import cgl.narada.service.fragmentation.CoalescingService;
import cgl.narada.service.fragmentation.FragmentationService;
import cgl.narada.service.order.OrderedDeliveryServiceImpl;
import cgl.narada.service.qos.ConsumerConstraints;
import cgl.narada.service.qos.ProducerConstraints;
import cgl.narada.service.qos.QosEvent;
import cgl.narada.service.qos.QosIdentifiers;
import cgl.narada.service.reliable.events.RdArchivalServiceNotification;
import cgl.narada.service.reliable.events.RdRecoveryEntityRequest;
import cgl.narada.service.reliable.events.ReliableDeliveryExchange;
import cgl.narada.service.reliable.impl.RdEntityActuatorImpl;
import cgl.narada.service.reliable.impl.RdPublishingEntityImpl;
import cgl.narada.service.security.SecureTopicsManager;
import cgl.narada.service.security.impl.EntityOperationsImpl;
import cgl.narada.service.security.kmc.SignedSecurityToken;
import cgl.narada.service.security.kmc.TSSGenerator;
import cgl.narada.service.security.kmc.TimeStampedSeqNo;
import cgl.narada.service.security.securityprovider.CertificateUtil;
import cgl.narada.service.tds.TDService;
import cgl.narada.service.tds.TDServiceImpl;
import cgl.narada.service.time.ntptime.NtpTimeService;
import cgl.narada.service.timer.ClockI;
import cgl.narada.service.timer.HRClock;

/**
 * This class provides a set of core functions that service implementations need
 * to satisfy. All services within the system need to conform to this interface.
 * 
 * @author Shrideep Pallickara $Date$ $Revision$
 */

public class QosServiceImpl {

    // implements ReliableDeliveryServiceListener,
    // OrderedDeliveryServiceListener {

    /**
     * This variable is used to assign unique identifiers for consumers and
     * producers.
     */
    private int generator = 1618;

    private NtpTimeService ntpTimeService;

    private EntityProfileMatchingTrees entityMatchingTrees;

    /** Keeps track of all the registered producers and consumers */
    private Hashtable consumers, producers;

    private CommunicationsService communicationsService;

    private int numOfConsumers = 0;

    private int numOfProducers = 0;

    /** Keeps a list of the consumers associated with profiles */
    private Hashtable profilesAndConsumers;

    /** keeps a list of the constraints associated with a given profile */
    private Hashtable profilesAndConstraints;

    private Hashtable recoveryInfo;

    /** For testing purposes */
    private int tracker = 0;

    private boolean toggle = false;

    private RdPublishingEntityImpl rdPublishingEntity;

    private RdEntityActuatorImpl entityActuator;

    private boolean connectionInitialized = false;

    private NBEventGenerator eventGenerator;

    private FragmentationService fragmentationService;

    private CoalescingService coalescingService;

    private CompressionService compressionService;

    private DecompressionService decompressionService;

    private OrderedDeliveryServiceImpl orderedDeliveryService;

    private BufferingServiceImpl bufferingService;

    private TDService timeDifferentialService;

    // HG: Added to store last timestamp and seq. id
    // private static Hashtable securityTimeStampManager = new Hashtable();

    private boolean debug = false;

    private static boolean timeServiceStarted = false;

    private int entityId;

    private static boolean tdsServicesStarted = false;

    private ServiceProperties serviceProperties;

    private String moduleName = "QosServiceImpl: ";

    // HG: Benchmark..
    boolean benchmark = false;

    BenchmarkListener benchmarkListener;

    ClockI hrClock = null;

    // HG: Communication Loss Listener...
    private ConnectionLossListener connLossListener;

    public QosServiceImpl(ServiceProperties serviceProperties) {
        this.serviceProperties = serviceProperties;
        // ServiceProperties.getServiceProperties().initialize(serviceProperties);
        cgl.narada.service.replay.EntityReplayActuator ea = cgl.narada.service.replay.impl.EntityReplayActuatorImpl
            .getInstance();
        ea.setQosService(this);
        ntpTimeService = NtpTimeService.getInstance();

        if (!timeServiceStarted) {
            timeServiceStarted = true;
            ntpTimeService.setQosServiceImpl(this);
            ntpTimeService.start();

        }

        consumers = new Hashtable();
        producers = new Hashtable();

        entityMatchingTrees = new EntityProfileMatchingTrees();

        communicationsService = new CommunicationsService(this);
        communicationsService.start();

        profilesAndConsumers = new Hashtable();

        profilesAndConstraints = new Hashtable();

        entityActuator = new RdEntityActuatorImpl(this);

        recoveryInfo = new Hashtable();

        eventGenerator = new NBEventGenerator();

        orderedDeliveryService = new OrderedDeliveryServiceImpl(this);
        bufferingService = new BufferingServiceImpl(this);
        bufferingService.startBufferingService();
        fragmentationService = new FragmentationService(this);
        coalescingService = new CoalescingService(this);

        compressionService = new CompressionService(this);
        decompressionService = new DecompressionService(this);
        timeDifferentialService = new TDServiceImpl(this);

        // BENCHMARK
        if (benchmark) {
            benchmarkListener = BenchmarkListener.getInstance();
            benchmarkListener.register("QosEncrypt.out");
            benchmarkListener.register("QosDecrypt.out");
            hrClock = HRClock.getClock();
        }
    }

    public void startTdsServices() {
        timeDifferentialService.startTDServices();
        tdsServicesStarted = true;
    }

    public ServiceProperties getServiceProperties() {
        return serviceProperties;
    }

    public void restartRtspServices() {
        bufferingService.stopBufferingService();
        bufferingService.terminateService();
        timeDifferentialService.shutDownTDServices();

        bufferingService = null;
        bufferingService = new BufferingServiceImpl(this);

        timeDifferentialService = null;
        timeDifferentialService = new TDServiceImpl(this);
        bufferingService.startBufferingService();
        timeDifferentialService.startTDServices();
    }

    public TDService getTDService() {
        return timeDifferentialService;
    }

    public void connectionInitialized() {
        connectionInitialized = true;
    }

    public void initializeBrokerCommunications(int entityId, Properties props,
                                               String commType)
        throws ServiceException {
        communicationsService.loadCommunicationsOfType(props, commType);
        communicationsService.setupLink(props, commType);
        if (commType.equals("ipsec")) {
            System.out
                .println(moduleName + "Will not register entity to IPSec");
            return;
        } else {
            System.out.println(moduleName + "Registering entity [" + entityId
                + "] using the (" + commType + ") transport");
        }
        System.out.println(moduleName + "Registering using (" + commType + ")");
        communicationsService.registerEntity(entityId);
        long sleepTime = 0;
        long sleepInterval = 20;
        while (!connectionInitialized) {
            try {
                Thread.sleep(sleepInterval);
                sleepTime += sleepInterval;
            } catch (InterruptedException ioe) {
                System.out.println(moduleName
                    + "Problems sleeping while waiting for"
                    + "initialization of " + entityId);
            }

            System.out.print(".");
            if (sleepTime > 8000) {
                String errorReport = moduleName
                    + "[3] seconds have elapsed since connection attempt was made. "
                    + "The broker must be very busy. Please try again";
                throw new ServiceException(errorReport);
            }
        }

        rdPublishingEntity = new RdPublishingEntityImpl(this,
                                                        communicationsService);
        this.entityId = entityId;
    }

    public void closeBrokerConnection() throws ServiceException {
        communicationsService.closeConnection();
    }

    public boolean hasPendingTransfers() {
        return communicationsService.hasPendingTransfers();
    }

    // HG: Connection Loss Listener...

    public final void setConnLossListener(
                                          ConnectionLossListener connLossListener) {
        this.connLossListener = connLossListener;
    }

    public void processConnectionLoss() {
        if (connLossListener != null)
            connLossListener.handleConnectionLoss();
    }

    // ----

    public void terminateServices() throws ServiceException {
        communicationsService.terminateService();
        ntpTimeService.terminateService();
        bufferingService.terminateService();
        rdPublishingEntity.terminateService();
    }

    public int getEntityId() {
        return entityId;
    }

    public NBEventGenerator getEventGenerator() {
        return eventGenerator;
    }

    public FragmentationService getFragmentationService() {
        return fragmentationService;
    }

    public void processReceivedEvent(NBEvent nbEvent) {
        Hashtable dests = null;
        dests = entityMatchingTrees.computeEventDestinations(nbEvent);
        if (debug)
            System.out.println(moduleName + " Found " + dests.size()
                + " destinations !!");

        if (dests == null) {
            System.out.println(moduleName + "NO Matching destinations ");
            return;
        }

        if (dests.size() == 0) {
            System.out.println(moduleName + "NO Matching destinations ");
            return;
        }

        try {
            if (isArchivalNotification(nbEvent)) {
                if (debug) {
                    System.out.println("ARCHIVAL NOTIFICATION !!");
                }
                manageArchivalNotification(nbEvent, dests);
                return;
            }
        } catch (ServiceException serEx) {
            System.out.println(serEx);
        }

        manageEventRoutingToDestinations(nbEvent, dests);
    }

    private boolean isArchivalNotification(NBEvent nbEvent) {
        int eventType = nbEvent.getEventType();
        if (eventType != NBEventTypes.RELIABLE_DELIVERY_EXCHANGE) {
            return false;
        }

        byte[] contentPayload = nbEvent.getContentPayload();
        byte exchangeType = contentPayload[0];

        if (exchangeType == ReliableDeliveryExchange.ARCHIVAL_NOTIFICATION) {
            return true;
        }
        return false;
    }

    private void manageArchivalNotification(NBEvent nbEvent,
                                            Hashtable destinations)
        throws ServiceException {
        Vector uniqueEntityList = new Vector();

        Enumeration _consumers = destinations.keys();

        while (_consumers.hasMoreElements()) {
            Object _consumerKey = _consumers.nextElement();
            int _entityId = ((EventConsumer) consumers.get(_consumerKey))
                .getEntityId();
            Object _entityKey = new Integer(_entityId);

            if (!uniqueEntityList.contains(_entityKey)) {
                uniqueEntityList.addElement(_entityKey);
            }
        }

        byte[] contentPayload = nbEvent.getContentPayload();
        byte exchangeType = contentPayload[0];

        if (exchangeType == ReliableDeliveryExchange.ARCHIVAL_NOTIFICATION) {
            RdArchivalServiceNotification archNotification = new RdArchivalServiceNotification(
                                                                                               contentPayload);
            for (int i = 0; i < uniqueEntityList.size(); i++) {
                Object toGoTo = uniqueEntityList.elementAt(i);
                entityActuator
                    .processArchivalNotificationForEntity(toGoTo,
                                                          archNotification);
            }

            return;
        }

        System.out.println(moduleName + "Unknown Reliable Delivery Exchange ["
            + exchangeType + "] received");

    }

    private void manageEventRoutingToDestinations(NBEvent nbEvent,
                                                  Hashtable destinations) {
        Enumeration _profiles = destinations.elements();
        QosEvent qosEvent = null;
        while (_profiles.hasMoreElements()) {
            Profile _profile = (Profile) _profiles.nextElement();
            String _profileId = _profile.getProfileId();

            if (debug) {
                System.out.println(moduleName + "Computed match for profile "
                    + _profile);
            }
            if (profilesAndConstraints.containsKey(_profileId)) {
                if (qosEvent == null) {
                    qosEvent = new QosEventImpl(nbEvent);
                }

                ConsumerConstraints _consumerConstraints = (ConsumerConstraints) profilesAndConstraints
                    .get(_profileId);
                qosEvent.addConsumerConstraints(_consumerConstraints);

                if (debug) {
                    System.out.println(moduleName
                        + "Added consumer constraints for " + _profile
                        + " QosEvent");
                }
            } else {
                if (debug) {
                    System.out.println(moduleName
                        + "NO consumer constraints for profile" + _profile);
                }
            }

        }
        /** if profile has more elements */

        if (qosEvent == null) {
            if (debug) {
                System.out.println(moduleName + "NO QosEvent was created");
            }
            Enumeration _consumerDests = destinations.keys();
            while (_consumerDests.hasMoreElements()) {
                Object _consumerId = _consumerDests.nextElement();
                EventConsumer _consumer = (EventConsumer) consumers
                    .get(_consumerId);

                // NOTE, NO Consumer constraints were found !!, hence NULL !
                routeToEventConsumer(_consumer, null, nbEvent);
            }
        } else {
            manageConsumptionOfQosEvent(qosEvent);
        }

    }

    /**
     * The newly created QosEvent needs to be handled and processed to determine
     * if all constraints have been satisfied
     */
    // private void OldmanageConsumptionOfQosEvent(QosEvent qosEvent) {
    // if (debug) {
    // System.out.println(moduleName
    // + "Trying to manage consumption of QosEvent");
    // System.out.println(moduleName
    // + "Total number of specified constraints= "
    // + qosEvent.getNumberofConsumerConstraints());
    // }
    //
    // ConsumerConstraintsImpl _satisfiedConstraints = null;
    //
    // _satisfiedConstraints = (ConsumerConstraintsImpl) qosEvent
    // .getSatisfiedConsumerConstraints();
    // while (_satisfiedConstraints != null) {
    // EventConsumer _eventConsumer = (EventConsumer) consumers
    // .get(new Integer(_satisfiedConstraints.getConsumerId()));
    //
    // if (_satisfiedConstraints.isReceiveAfterCoalescingFragments()) {
    // if (qosEvent.getNBEvent().getEventHeaders().isFragmented()) {
    // try {
    // coalescingService.manageFragmentedEvent(_eventConsumer,
    // qosEvent.getNBEvent());
    // } catch (ServiceException serEx) {
    // System.out.println(moduleName + serEx);
    // } /* end try-catch */
    // return;
    // }
    // } else {
    // if (debug) {
    // System.out.println(moduleName
    // + "Routing to event consumer "
    // + _eventConsumer.getEntityId());
    // }
    // routeToEventConsumer(_eventConsumer, qosEvent.getNBEvent());
    // }
    // _satisfiedConstraints = (ConsumerConstraintsImpl) qosEvent
    // .getSatisfiedConsumerConstraints();
    // }
    //
    // // int serviceType = qosEvent.getServiceToBeRoutedTo();
    //
    // try {
    // if (qosEvent.getPendingConsumerConstraints() != null) {
    // entityActuator.processConstructedQosEvent(qosEvent);
    // }
    // } catch (ServiceException serEx) {
    // System.out.println(serEx);
    // }
    //
    // }
    /**
     * The newly created QosEvent needs to be handled and processed to determine
     * if all constraints have been satisfied
     */
    private void manageConsumptionOfQosEvent(QosEvent qosEvent) {
        if (debug) {
            System.out.println(moduleName
                + "Trying to manage consumption of QosEvent");
            System.out.println(moduleName
                + "Total number of specified constraints= "
                + qosEvent.getNumberofConsumerConstraints());
        }

        ConsumerConstraintsImpl _satisfiedConstraints = null;

        _satisfiedConstraints = (ConsumerConstraintsImpl) qosEvent
            .getSatisfiedConsumerConstraints();
        while (_satisfiedConstraints != null) {
            EventConsumer _eventConsumer = (EventConsumer) consumers
                .get(new Integer(_satisfiedConstraints.getConsumerId()));

            if (_satisfiedConstraints.isReceiveAfterCoalescingFragments()) {
                if (qosEvent.getNBEvent().getEventHeaders().isFragmented()) {
                    try {
                        coalescingService
                            .manageFragmentedEvent(_eventConsumer, qosEvent
                                .getNBEvent());
                    } catch (ServiceException serEx) {
                        System.out.println(moduleName + serEx);
                    } /* end try-catch */
                    return;
                }
            } else {
                if (debug) {
                    System.out.println(moduleName
                        + "Routing to event consumer "
                        + _eventConsumer.getEntityId());
                }

                routeToEventConsumer(_eventConsumer, _satisfiedConstraints,
                                     qosEvent.getNBEvent());
            }
            _satisfiedConstraints = (ConsumerConstraintsImpl) qosEvent
                .getSatisfiedConsumerConstraints();
        }

        // int serviceType = qosEvent.getServiceToBeRoutedTo();

        try {
            ConsumerConstraints[] _pendingConstraints = null;
            _pendingConstraints = qosEvent.getPendingConsumerConstraints();
            QosIdentifiers qosIdentifiers = qosEvent.getQosIdentifiers();

            if (_pendingConstraints == null) {
                if (debug) {
                    System.out.println(moduleName
                        + "Pending constraints == NULL!");
                }
                return;
            }

            for (int i = 0; i < _pendingConstraints.length; i++) {

                if (_pendingConstraints[i].isReceiveReliably()
                    && !qosIdentifiers.isArchived()) {
                    System.out.println(moduleName
                        + "Routing to reliable delivery actuator");
                    entityActuator.processConstructedQosEvent(qosEvent);
                    continue;
                }

                if (_pendingConstraints[i].isReceiveInTimeOrder()
                    && !qosIdentifiers.isTimeOrdered()) {
                    if (debug) {
                        System.out.println(moduleName
                            + "Routing to buffering service ");
                    }
                    bufferingService.timeOrderEvent(qosEvent);
                    continue;
                }

                if (_pendingConstraints[i].isReceiveAfterTimeSpacing()
                    && !qosIdentifiers.isTimeSpaced()) {

                    if (!tdsServicesStarted) {
                        startTdsServices();
                    }

                    if (debug) {
                        System.out.println(moduleName
                            + "Routing to time differential service");
                    }
                    timeDifferentialService.timeSpaceEvent(qosEvent);
                    continue;
                }

            }/* end for _pendingConstraints.length */
        } catch (ServiceException serEx) {
            System.out.println(serEx);
        }
    }

    private void routeToEventConsumer(EventConsumer consumer,
                                      ConsumerConstraints consumerConstraints,
                                      NBEvent nbEvent) {

        long start = 0, end = 0;

        try {

            if (nbEvent.hasEventHeaders()) {

                EventHeaders hdr = nbEvent.getEventHeaders();

                // HG: Security suff...
                // If its a secure event then decode the payload before
                // final delivery

                // Step 1: Check if its a secure event AND secure delivery was
                // requested
                if (consumerConstraints != null) {
                    if (hdr.isSecure()
                        && consumerConstraints.isReceiveSecurely()) {

                        // BENCHMARK
                        if (benchmark) {
                            start = hrClock.getTimeMicroseconds();
                        }

                        PublicKey rootCAPublicKey = consumerConstraints
                            .getRootCAPublicKey();

                        // Step 2: Verify payload, Check the signature ??

                        if (!hdr.hasSignature()) {
                            System.err.println("No security token found !");
                            return;
                        }

                        // Verify the Signed security token's certificate
                        SignedSecurityToken sst = SignedSecurityToken
                            .createObjectFromBytes(hdr.getSignedSecurityToken());

                        // Step 1: Verify the certificate with Root CA's public
                        // key
                        Certificate cert = sst.getKmcCertificate();
                        if (!CertificateUtil
                            .ValidateCertificate(cert, rootCAPublicKey)) {
                            System.err
                                .println("KMC's certificate could not be validated !");
                            return;
                        }

                        // Step 2: Now check the payload's signature
                        // Step 2a: Get the entity's certificate from the
                        // certificate's table
                        Certificate entityCert = sst.getSecurityToken()
                            .getTopicOwnerCertificate();

                        // BigInteger certId = ((X509Certificate)
                        // entityCert).getSerialNumber();
                        //
                        // // Check if the certificate is in the revoked list
                        // if (revokedCertificates.contains(certId)) {
                        // log.error("Entity's certificate has been revoked !");
                        // return false;
                        // }

                        // Verify the certificate
                        if (!CertificateUtil
                            .ValidateCertificate(entityCert, rootCAPublicKey)) {
                            System.err
                                .println("Entity's certificate could not be validated !");
                            return;
                        }

                        // Validate the Signed Security token
                        if (!sst.verify()) {
                            System.err
                                .println("Could not verify integrity of signed security token !");
                            return;
                        }

                        // BENCHMARK
                        if (benchmark) {
                            end = hrClock.getTimeMicroseconds();
                            benchmarkListener
                                .set("QosDecrypt.out",
                                     ("validateSST: " + (end - start)));
                            start = hrClock.getTimeMicroseconds();
                        }

                        // Check the signature
                        try {
                            EntityOperationsImpl entityOperations = new EntityOperationsImpl();

                            // Check Header signature
                            byte[] sig;
                            try {
                                sig = (byte[]) nbEvent
                                    .getEventProperties()
                                    .getProperty(
                                                 SecureTopicsManager.HEADER_SIGNATURE);
                            } catch (NBEventException e) {
                                sig = null;
                            }

                            if (sig == null) {
                                System.err
                                    .println("Header signature not found !");
                                return;
                            }

                            String signingAlgorithm = hdr
                                .getSignatureAlgorithm();

                            if (!entityOperations.validateSignature(entityCert
                                .getPublicKey(), signingAlgorithm, "BC", hdr
                                .getBytes(), sig)) {
                                System.err
                                    .println("Could not validate Event header signature !");
                                return;
                            }

                            // Check payload signature
                            if (!entityOperations
                                .validateSignature(entityCert.getPublicKey(),
                                                   signingAlgorithm, "BC",
                                                   nbEvent.getContentPayload(),
                                                   hdr.getSignature())) {
                                System.err
                                    .println("Could not validate Event payload signature !");
                                return;
                            }
                        } catch (ServiceException e) {
                            System.err.println(e.getMessage());
                            return;
                        }

                        // BENCHMARK
                        if (benchmark) {
                            end = hrClock.getTimeMicroseconds();
                            benchmarkListener
                                .set("QosDecrypt.out",
                                     ("validateSignature: " + (end - start)));
                        }

                        // Ok, Everything checks FINE... this event is good to
                        // go !

                        // Step 2: By default MUST decrypt, but client MAY
                        // specify NO decryption

                        if (!consumerConstraints.isNoDecryptionRequested()) {

                            // OK decrypt the message
                            SecretKey secretKey = consumerConstraints
                                .getSecretKey();
                            Properties decryptionProps = consumerConstraints
                                .getDecryptionProperties();

                            if (secretKey == null) {
                                // No security token found, so discard...
                                System.err
                                    .println("OOPS... Decryption specified "
                                        + "BUT No Secret key specified! "
                                        + "Discarding EVENT...");
                                return;
                            } else {
                                // System.out.println("Decrypting !");

                                byte[] originalPayload = nbEvent
                                    .getContentPayload();

                                byte[] iv = hdr.getCipherIV();
                                String mode = hdr.getCipherMode();
                                String padding = hdr.getCipherPaddingScheme();

                                String algModePadding = secretKey
                                    .getAlgorithm()
                                    + "/" + mode + "/" + padding;

                                EntityOperationsImpl eoi = new EntityOperationsImpl();

                                // BENCHMARK
                                if (benchmark) {
                                    start = hrClock.getTimeMicroseconds();
                                }

                                byte[] decryptedPayload = eoi
                                    .decryptPayload(secretKey, originalPayload,
                                                    algModePadding, iv, "BC");

                                // BENCHMARK
                                if (benchmark) {
                                    end = hrClock.getTimeMicroseconds();
                                    benchmarkListener
                                        .set("QosDecrypt.out",
                                             ("Decrypt: " + (end - start)));
                                }

                                this.eventGenerator
                                    .updateWithDecryptedPayload(nbEvent,
                                                                decryptedPayload);

                            }
                        }
                    } else {
                        // If NO consumer constraints and a secure event is
                        // recieved..
                        // OR
                        // If consumer constraints for secure delivery but not a
                        // secure event
                        // THEN drop it
                        if (hdr.isSecure()
                            || consumerConstraints.isReceiveSecurely()) {
                            System.out
                                .println("ERROR: Secure Event recieved BUT "
                                    + "no matching secure reciever constraints !");
                            return;
                        }
                    }
                } else {
                    // If no consumer constraints and a secure event is
                    // recieved.. then drop it
                    if (hdr.isSecure())
                        System.out.println("ERROR: Secure Event recieved BUT "
                            + "no matching secure reciever constraints !");
                }
                // ---

                if (hdr.isCompressed()) {
                    decompressionService.decompressEvent(nbEvent);

                    // WHY THIS RETURN HERE ?
                    return;
                }

            }/* end if headers */

            if (debug) {
                System.out.println(moduleName + "Routing to consumer ["
                    + +consumer.getEntityId() + "]");
            }

            consumer.getNBEventListener().onEvent(nbEvent);

        } catch (ServiceException serEx) {
            System.out.println(moduleName + serEx);
        } /* end try-catch */
    }

    public void onReliableDelivery(
                                   QosEvent qosEvent,
                                   RdArchivalServiceNotification archNotification) {
        QosIdentifiersImpl qosIdentifiers = (QosIdentifiersImpl) qosEvent
            .getQosIdentifiers();
        qosIdentifiers.setArchived();
        if (debug) {
            System.out.println(moduleName + "Event now reliable ...");
            System.out.println(moduleName
                + "Total number of specified constraints= "
                + qosEvent.getNumberofConsumerConstraints()
                + " entering onReliableDelivery()");
        }

        /** This will be used to ensure guaranteed exactly once ordered delivery */
        ConsumerConstraints[] _pendingConstraints = qosEvent
            .getPendingConsumerConstraints();
        for (int i = 0; i < _pendingConstraints.length; i++) {

            if (_pendingConstraints[i].isReceiveInOrder()) {
                System.out.println("\n\n" + moduleName
                    + "Getting into ordered " + "delivery service \n\n");
                orderedDeliveryService.onReliableDelivery(qosEvent,
                                                          archNotification);
                break;
            }/* end if */
        }
        /** end of processing related to exactly once delivery */

        manageConsumptionOfQosEvent(qosEvent);
    }

    public void onAdvanceOfSyncpoint(int entityId, int templateId,
                                     long advanceSync) {
        orderedDeliveryService.onAdvanceOfSyncpoint(entityId, templateId,
                                                    advanceSync);
    }

    public void onOrderedDelivery(QosEvent qosEvent) {
        QosIdentifiersImpl qosIdentifiers = (QosIdentifiersImpl) qosEvent
            .getQosIdentifiers();
        qosIdentifiers.setOrdered();
        manageConsumptionOfQosEvent(qosEvent);
    }

    public void onTimeOrderedDelivery(QosEvent qosEvent) {
        QosIdentifiersImpl qosIdentifiers = (QosIdentifiersImpl) qosEvent
            .getQosIdentifiers();
        qosIdentifiers.setTimeOrdered();
        manageConsumptionOfQosEvent(qosEvent);
    }

    public void onTimeSpacedDelivery(QosEvent qosEvent) {
        QosIdentifiersImpl qosIdentifiers = (QosIdentifiersImpl) qosEvent
            .getQosIdentifiers();
        qosIdentifiers.setTimeSpaced();
        System.out.println(moduleName + "Event time-spaced by TDS");
        manageConsumptionOfQosEvent(qosEvent);
    }

    public int generateConsumerId() {
        return generator++;
    }

    public int generateProducerId() {
        return generator++;
    }

    public int generateConstraintId() {
        return generator++;
    }

    /** This method registers a consumer with the Qos manager */
    public void registerConsumer(int consumerId, EventConsumer consumer) {
        Integer key = new Integer(consumerId);

        if (consumers.containsKey(key)) {
            System.out.println(moduleName + "ConsumerId [" + consumerId
                + "] exists. This should not be the case! ");
            return;
        }

        consumers.put(key, consumer);
        numOfConsumers++;
    }

    public void deregisterConsumer(int consumerId) {
        Integer key = new Integer(consumerId);
        if (!consumers.containsKey(key)) {
            System.out.println(moduleName + "ConsumerId [" + consumerId
                + "] does not exist. This should not be the case! ");
            return;
        }

        consumers.remove(key);
        numOfConsumers--;
    }

    /** This method is used to register producers with the QoS Manager. */
    public void registerProducer(int producerId, EventProducer producer) {
        Integer key = new Integer(producerId);

        if (producers.containsKey(key)) {
            System.out.println(moduleName + "ProducerId [" + producerId
                + "] exists. This should not be the case! ");
            return;
        }

        producers.put(key, producer);
        numOfProducers++;
    }

    /** This method is used to register producers with the QoS Manager. */
    public void deregisterProducer(int producerId) {
        Integer key = new Integer(producerId);

        if (!producers.containsKey(key)) {
            System.out.println(moduleName + "ProducerId [" + producerId
                + "] does not exist. This should not be the case! ");
            return;
        }

        producers.remove(key);
        numOfProducers--;
    }

    public void publishEvent(NBEvent nbEvent,
                             ProducerConstraints producerConstraints,
                             int producerId) throws ServiceException {

        if (toggle && producerConstraints == null) {
            tracker++;
            if (tracker % 5 == 0) {
                System.out.println(moduleName + "Skipping publish for tracker "
                    + tracker);
                return;
            }
        }

        if (producerConstraints == null) {
            communicationsService.sendData(nbEvent.getBytes());
            return;
        }

        // toggle = true;
        if (producerConstraints.isSendAfterFragmentation()) {
            Properties fragmentationProperties = producerConstraints
                .getFragmentationParams();
            fragmentationService.fragmentEvent(nbEvent, producerConstraints);
            return;
        }

        if (producerConstraints.isSendAfterPayloadCompression()) {
            Properties compressionProperties = producerConstraints
                .getCompressionParams();
            compressionService.compressEvent(nbEvent, compressionProperties,
                                             true);
            // HG: Return removed...
            // return;
        }

        // HG: Security stuff...
        // NOTE: We ENCRYPT before fragmentation and compression...
        // MUST maintain similar order...
        // ==> fragment -> compress -> encrypt -> SEND
        // ==> decompress...-> Defragment -> decrypt => DELIVER

        if (producerConstraints.isSendSecurely()) {

            // Validate if the security token is still valid...
            SignedSecurityToken sst = producerConstraints
                .getSignedSecurityToken();

            long start = 0, t1 = 0, t2 = 0, t3 = 0;

            // Step 1: Encrypt the message
            SecretKey secretKey = producerConstraints.getSecretKey();

            // Set the default properties...
            Properties encryptionProps = producerConstraints
                .getEncryptionProperties();

            String mode = "CBC", padding = "PKCS7Padding", signingAlgorithm = "SHA1withRSA";
            if (encryptionProps != null) {
                mode = encryptionProps
                    .getProperty(ProducerConstraints.CIPHER_MODE);
                if (mode == null)
                    mode = "CBC";

                padding = encryptionProps
                    .getProperty(ProducerConstraints.CIPHER_PADDING);

                if (padding == null)
                    padding = "PKCS7Padding";

                signingAlgorithm = encryptionProps
                    .getProperty(ProducerConstraints.SIGNING_ALGORITHM);

                if (signingAlgorithm == null)
                    signingAlgorithm = "SHA1withRSA";

                if (debug)
                    System.out.println("Using: mode=" + mode + ", padding="
                        + padding + ", sigAlg=" + signingAlgorithm);
            }

            byte[] originalPayload = nbEvent.getContentPayload();

            EventHeaders headers = nbEvent.getEventHeaders();

            // System.out.println("Encrypting !");

            EntityOperationsImpl eoi = new EntityOperationsImpl();

            String algModePadding = secretKey.getAlgorithm() + "/" + mode + "/"
                + padding;

            byte[] iv = eoi.getIV(secretKey.getAlgorithm(), mode, padding);

            // BENCHMARK
            if (benchmark) {
                start = hrClock.getTimeMicroseconds();
            }

            byte[] encryptedPayload = eoi.encryptPayload(secretKey,
                                                         originalPayload,
                                                         algModePadding, iv,
                                                         "BC");

            // BENCHMARK [Encryption]
            if (benchmark) {
                t1 = hrClock.getTimeMicroseconds();
            }

            // Step 2: Sign the encrypted payload
            byte[] signature = producerConstraints
                .signPayload(encryptedPayload, signingAlgorithm);

            // replace the content payload
            this.eventGenerator.updateWithEncryptedPayload(nbEvent,
                                                           encryptedPayload);

            // BENCHMARK [Signing Encrypted payload]
            if (benchmark) {
                t2 = hrClock.getTimeMicroseconds();
            }

            // Set the security token and signature.
            headers.setEncryptedPayloadInformation(iv, mode, padding);
            headers.setSecurityToken(sst.getBytes());
            headers.setSignatureInformation(signingAlgorithm, signature);

            // Finally Add a timestamped seq No.
            // NOTE: The stssn MUST always be monotonically increasing value
            TimeStampedSeqNo stssn = TSSGenerator.generate();

            if (stssn == null) {
                System.err
                    .println("Unable to get timestamp, Cannot route packet !");
                return;
            }

            headers.setSecurityTimeStampedSequenceNo(stssn.getBytes());

            // Sign the Event header and include that
            byte[] headerSignature = producerConstraints.signPayload(headers
                .getBytes(), signingAlgorithm);
            try {
                nbEvent.getEventProperties()
                    .setImmutableProperty(SecureTopicsManager.HEADER_SIGNATURE,
                                          headerSignature,
                                          QosServiceImpl.class.getName());
            } catch (NBEventException e) {
                System.err.println("Unable to set IMMUTABLE property: "
                    + SecureTopicsManager.HEADER_SIGNATURE);
                System.err.println("Error: " + e);
                return;
            }

            // BENCHMARK [Marshall and sign header]
            if (benchmark) {
                t3 = hrClock.getTimeMicroseconds();

                benchmarkListener.set("QosEncrypt.out",
                                      ("EncryptPayload: " + (t1 - start)));

                benchmarkListener.set("QosEncrypt.out",
                                      ("SignPayload: " + (t2 - t1)));
                benchmarkListener.set("QosEncrypt.out",
                                      ("MarshalAndSignHeader: " + (t3 - t2)));
            }
        }

        // Check if reliable delivery is requested...
        if (!producerConstraints.isSendReliably()) {
            // Just send whatever we have until now...
            communicationsService.sendData(nbEvent.getBytes());
            return; // return added here...
        }
        // ---

        if (debug) {
            System.out.println(moduleName + "Publishing reliably ..");
        }

        QosEvent qosEvent = new QosEventImpl(nbEvent, producerConstraints);
        rdPublishingEntity.publishReliably(qosEvent);

    }

    public void subscribeTo(Profile profile,
                            ConsumerConstraints consumerConstraints,
                            int consumerId) throws ServiceException {
        entityMatchingTrees
            .manageSubscriptionProfile(profile, true, consumerId);

        ProfileRequest profileRequest = new ProfileRequest(profile, true);

        // HG: If security token was specified in consumer constraints...then
        // include it in the profile request
        if (consumerConstraints != null) {
            System.out
                .println("*** -> Found ConsumerConstraints: checking for secure token");

            if (consumerConstraints.isReceiveSecurely()) {
                System.out.println("*** -> Found Secure delivery request");
                SignedSecurityToken sst = consumerConstraints
                    .getSignedSecurityToken();

                System.out.println("SST = " + (sst != null));
                profileRequest.setSignedSecurityToken(sst.getBytes());
            }
        }
        // ---

        // Send profile connection request to broker !
        communicationsService.sendData(profileRequest.getBytes());

        Object profileKey = profile.getProfileId();
        profilesAndConsumers.put(profileKey, new Integer(consumerId));

        if (debug) {
            System.out.println(moduleName + "Registering profile with Id -> "
                + profileKey);
        }

        if (consumerConstraints != null) {
            System.out.println(moduleName
                + "Registering consumer constraints for " + profile);
            profilesAndConstraints.put(profileKey, consumerConstraints);
            if (consumerConstraints.isReceiveReliably()) {
                entityActuator.issueProfileUpdateRequest(profile, true,
                                                         consumerConstraints);
            } /* isReceiveReliably */
        }/* end consumerConstraints != null */
    }

    public void unsubscribe(Profile profile) throws ServiceException {

        Object profileKey = profile.getProfileId();
        if (!profilesAndConsumers.containsKey(profileKey)) {
            throw new ServiceException(moduleName + "Profile with id ["
                + profileKey + "] not previously registered");
        }

        Integer consumerInt = (Integer) profilesAndConsumers.get(profileKey);
        int consumerId = consumerInt.intValue();

        entityMatchingTrees.manageSubscriptionProfile(profile, false,
                                                      consumerId);

        profilesAndConsumers.remove(profileKey);
        profilesAndConstraints.remove(profileKey);

        ProfileRequest profileRequest = new ProfileRequest(profile, false);
        communicationsService.sendData(profileRequest.getBytes());

    }

    public long recoverEntity(int conducerId, int entityId, int templateId,
                              boolean isPublisher) throws ServiceException {
        Object recoveryKey = new Long(System.currentTimeMillis());

        while (recoveryInfo.containsKey(recoveryKey)) {
            /**
             * This ensures that every recovery has a different request
             * associated with it
             */
            recoveryKey = new Long(System.currentTimeMillis());
        }
        long recoveryId = ((Long) recoveryKey).longValue();

        RdRecoveryEntityRequest recoveryRequest = new RdRecoveryEntityRequest(
                                                                              templateId,
                                                                              entityId,
                                                                              isPublisher,
                                                                              recoveryId);
        // communicationsService.sendData(recoveryRequest.getBytes());
        if (isPublisher) {
            rdPublishingEntity.issueRecoveryRequest(recoveryRequest);
        } else {
            entityActuator.issueRecoveryRequest(recoveryRequest);
        }
        recoveryInfo.put(recoveryKey, new Integer(conducerId));

        return recoveryId;

    }

    public EventConsumerImpl getRecoveringConsumer(long recoveryId) {
        Object recoveryKey = new Long(recoveryId);
        if (!recoveryInfo.containsKey(recoveryKey)) {
            System.out.println(moduleName + "Recovery with id " + recoveryId
                + " not known");
            return null;
        }

        Object consumerKey = recoveryInfo.get(recoveryKey);

        if (!consumers.containsKey(consumerKey)) {
            System.out.println(moduleName
                + "While recovering, consumer with id " + consumerKey
                + " not known");
            return null;
        }

        return (EventConsumerImpl) consumers.get(consumerKey);
    }

    public EventProducerImpl getRecoveringProducer(long recoveryId) {
        Object recoveryKey = new Long(recoveryId);
        if (!recoveryInfo.containsKey(recoveryKey)) {
            System.out.println(moduleName + "Recovery with id " + recoveryId
                + " not known");
            return null;
        }

        Object producerKey = recoveryInfo.get(recoveryKey);

        if (!producers.containsKey(producerKey)) {
            System.out.println(moduleName
                + "While recovering, producer with id " + producerKey
                + " not known");
            return null;
        }

        return (EventProducerImpl) producers.get(producerKey);
    }
}
